package com.fanxuankai.boot.confluent.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.extra.template.Template;
import cn.hutool.extra.template.TemplateEngine;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.fanxuankai.boot.confluent.dao.KsqldbObjectDao;
import com.fanxuankai.boot.confluent.domain.KsqldbObject;
import com.fanxuankai.boot.confluent.dto.KsqldbObjectDTO;
import com.fanxuankai.boot.confluent.dto.KsqldbObjectDropDTO;
import com.fanxuankai.boot.confluent.mapstruct.KsqldbObjectConverter;
import com.fanxuankai.boot.confluent.service.KsqldbObjectService;
import io.confluent.ksql.api.client.Client;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.io.StringWriter;
import java.util.Map;

/**
 * @author fanxuankai
 */
@Service
public class KsqldbObjectServiceImpl implements KsqldbObjectService {
    @Resource
    private KsqldbObjectDao dao;
    @Resource
    private KsqldbObjectConverter converter;
    @Resource
    private Client client;
    @Resource
    private TemplateEngine engine;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void create(KsqldbObjectDTO dto) throws Exception {
        if (dao.getOne(Wrappers.lambdaQuery(KsqldbObject.class).eq(KsqldbObject::getName, dto.getName())) != null) {
            return;
        }
        KsqldbObject ksqldbObject = converter.toEntity(dto);
        if (dto.isAutoCreate()) {
            ksqldbObject.setCreated(1);
        }
        dao.save(ksqldbObject);
        if (dto.isAutoCreate()) {
            Template template = engine.getTemplate("ksqldb_create_object.ftl");
            Map<?, ?> bindingMap = MapUtil.newHashMap();
            BeanUtil.copyProperties(ksqldbObject, bindingMap);
            StringWriter writer = new StringWriter();
            template.render(bindingMap, writer);
            String statement = writer.toString();
            client.executeStatement(statement).get();
        }
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void drop(KsqldbObjectDropDTO dto) throws Exception {
        KsqldbObject ksqldbObject = dao.getOne(Wrappers.lambdaQuery(KsqldbObject.class)
                .eq(KsqldbObject::getName, dto.getName()));
        if (ksqldbObject == null) {
            return;
        }
        dao.removeById(ksqldbObject.getId());
        Template template = engine.getTemplate("ksqldb_drop_object.ftl");
        Map<?, ?> bindingMap = MapUtil.newHashMap();
        BeanUtil.copyProperties(dto, bindingMap);
        StringWriter writer = new StringWriter();
        template.render(bindingMap, writer);
        String statement = writer.toString();
        client.executeStatement(statement).get();
    }
}
